package com.xq.jltx.learnflow

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlin.system.measureTimeMillis


/**
 * @author aidenquan
 *         Created by junlintianxia on 2021/08/11.
 */
fun main() = runBlocking<Unit> {
//    flowSimpleRun()
    flowSimpleCancelRun()
//    measureFlowTime()
//    channelFlowRun()
}

/**
 * flow 是 Cold Stream。在没有切换线程的情况下，生产者和消费者是同步非阻塞的。
 * channel 是 Hot Stream。而 channelFlow 实现了生产者和消费者异步非阻塞模型
 */
private suspend fun channelFlowRun() {
    val channelFlow = channelFlow<Int> {
        println("channelFlow started")
        (1..5).forEach {
            delay(100)
            println("channelFlow send $it thread: ${Thread.currentThread()}")
            send(it)
        }
    }
    println("channelFlow Calling collect...")
    channelFlow.collect {
        println("[1]$it thread: ${Thread.currentThread()}")
    }
    println("channelFlow Calling collect again...")
    channelFlow.collect {
        println("[2]$it thread: ${Thread.currentThread()}")
    }
    println("channelFlow Calling collect end...")
}

/**
 * 不切线程方式去获取flowTime执行时间
 */
private suspend fun measureFlowTime() {
    val time = measureTimeMillis {
        flow<Int> {
            println("Flow started")
            (1..5).forEach {
                delay(100)
                println("Flow emit $it thread: ${Thread.currentThread()}")
                emit(it)
            }
        }.collect {
            delay(100)
            println("$it thread: ${Thread.currentThread()}")
        }
    }
    println("time $time")
}


/**
 * 流是冷的
 * 一种类似于序列的冷流 — 这段 flow 构建器中的代码直到流被收集的时候才运行
 * simple() 调用会尽快返回且不会进行任何等待。该流在每次收集的时候启动，
 * 这就是为什么当我们再次调用 collect 时我们会看到“Flow started”
 */
private suspend fun CoroutineScope.flowSimpleRun() {
    launch(Dispatchers.IO) {
        for (k in 1..3) {
            println("I'm not blocked $k, ${Thread.currentThread()}")
            delay(1000)
        }
    }
    println("Calling simple function...")
    val flow = simpleFlow()
    println("Calling collect...")
    flow.collect {
        println("collect ${it}, ${Thread.currentThread()}")
    }
    println("Calling collect again...")
    flow.collect {
        println("collect1 ${it}, ${Thread.currentThread()}")
    }
    println("Calling collect end")
}

/**
 * 流取消
 *在运行的时候流是如何在超时的情况下取消并停止执行
 */
private suspend fun flowSimpleCancelRun() {
    withTimeoutOrNull(2500) {
        println("Calling collect...")
        simpleFlowCancel().collect {
            println(it)
        }
        println("Calling collect end")
    }
}

/**
 * 挂起函数
然而，计算过程阻塞运行该代码的主线程。
当这些值由异步代码计算时，我们可以使用 suspend 修饰符标记函数 simple，
这样它就可以在不阻塞的情况下执行其工作并将结果作为列表
使用 List 结果类型，意味着我们只能一次返回所有值
 */
suspend fun simpleList(): List<Int> {
    delay(1000) //假设做了一些异步的事情
    return listOf(1, 2, 3)
}

/**
 * 使用 List 结果类型，意味着我们只能一次返回所有值。 为了表示异步计算的值流（stream），我们可以使用 Flow 类型
 * 1.名为 flow 的 Flow 类型构建器函数。
 * 2.flow { ... } 构建块中的代码可以挂起。
 * 3.函数 simple 不再标有 suspend 修饰符。
 * 4.流使用 emit 函数 发射 值。
 * 5.流使用 collect 函数 收集 值。
 */
fun simpleFlow(): Flow<Int> = flow {
    println("Flow started")
    (1..3).forEach {
        delay(1000)
        emit(it)
    }
}

/**
 * 流取消
 * 流采用与协程同样的协作取消。像往常一样，流的收集可以在当流在一个可取消的挂起函数（例如 delay）中挂起的时候取消
 *
 */
fun simpleFlowCancel(): Flow<Int> = flow {
    println("Flow started")
    (1..5).forEach {
        delay(1000)
        println("Flow emit $it")
        emit(it)
    }
}
